package xuecheng.tool.mq.alibaba.rocket;

import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;

import javax.annotation.PreDestroy;

import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.SpringBootConfiguration;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.annotation.Bean;

import com.aliyun.openservices.ons.api.ONSFactory;
import com.aliyun.openservices.ons.api.Producer;
import com.aliyun.openservices.ons.api.PropertyKeyConst;

import lombok.NonNull;
import lombok.extern.slf4j.Slf4j;

/**
 * 自定义 RocketMq 配置
 *
 * @author PengXueCheng
 * @version 1.0.0
 */
@SpringBootConfiguration
@EnableConfigurationProperties(RocketMqProperties.class)
@ConditionalOnProperty(prefix = "cpx.rocketmq", value = "enable", havingValue = "true")
@Slf4j
public class CpxRocketMqAutoConfig {

    @Autowired
    private RocketMqProperties rocketMqProperties;

    @Bean
    public RocketMqUtils rocketMqUtils(Map<String, Producer> producers) {
        @NonNull
        String nameServer = rocketMqProperties.getNameServer();
        return new RocketMqUtils(producers, nameServer);
    }

    /**
     * 注册生产者
     */
    @Bean("producers")
    public Map<String, Producer> producers() {
        @NonNull
        String ak = rocketMqProperties.getAccessKey();
        @NonNull
        String sk = rocketMqProperties.getSecretKey();
        @NonNull
        String defaultNameServer = rocketMqProperties.getNameServer();
        @NonNull
        List<RocketMqProperties.Producer> producerConfigs = rocketMqProperties.getProducers();

        // 生产者集合
        Map<String, Producer> producers = new HashMap<>(producerConfigs.size());
        producerConfigs.stream().forEach(producerConfig -> {
            @NonNull
            String gid = producerConfig.getGroup();
            String nameServer = producerConfig.getNameServer();
            if (StringUtils.isEmpty(nameServer)) {
                nameServer = defaultNameServer;
            }

            Properties properties = new Properties();
            // 您在控制台创建的 Group ID
            properties.put(PropertyKeyConst.GROUP_ID, gid);
            // 鉴权用 AccessKey，在阿里云服务器管理控制台创建
            properties.put(PropertyKeyConst.AccessKey, ak);
            // 鉴权用 SecretKey，在阿里云服务器管理控制台创建
            properties.put(PropertyKeyConst.SecretKey, sk);
            // 设置 TCP 接入域名，进入控制台的实例管理页面，在页面上方选择实例后，在实例信息中的“获取接入点信息”区域查看
            properties.put(PropertyKeyConst.NAMESRV_ADDR, nameServer);
            Long sendMsgTimeoutMillis = producerConfig.getSendMsgTimeoutMillis();
            if (null != sendMsgTimeoutMillis) {
                properties.put(PropertyKeyConst.SendMsgTimeoutMillis, sendMsgTimeoutMillis);
            }

            Producer producer = ONSFactory.createProducer(properties);
            // 在发送消息前，必须调用 start 方法来启动 Producer，只需调用一次即可
            producer.start();

            String producerKey = String.format("%s-%s", nameServer, gid);
            log.info("Producer started. producerGroup: {}, nameServer: {}", gid, nameServer);
            // 注册所有生产者
            producers.put(producerKey, producer);
        });
        return producers;
    }

    /**
     * 销毁所有生产者
     */
    @PreDestroy
    public void shutdown(Map<String, Producer> producers) {
        // 在应用退出前，可以销毁 Producer 对象
        // 注意：如果不销毁也没有问题
        producers.values().stream().forEach(Producer::shutdown);
    }

}
